package com.need6

import ch.hsr.geohash.GeoHash
import com.need3.GetRedis
import com.test.Sk_Ak
import org.apache.spark.{SparkConf, SparkContext}
import com.typesafe.config.{Config, ConfigFactory}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SQLContext}
import redis.clients.jedis.Jedis

/**
  * Created by zhuang on 2018/3/7.
  */
object GetLog_Lat extends App {
  private val load: Config = ConfigFactory.load()
  val conf = new SparkConf().setMaster("local[*]").setAppName(this.getClass.getSimpleName)
    .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
  val sc = new SparkContext(conf)
  val context: SQLContext = new SQLContext(sc)
  //读取文件
  private val parquet: DataFrame = context.read.parquet(load.getString("DataForParquet"))
  parquet.foreachPartition(t => {
    val redis: Jedis = GetRedis.getRedis()
    t.foreach(row => {
      val log = row.getAs[String]("long")
      val lat = row.getAs[String]("lat")
      if (lat.size != 0 && log.size != 0) {
        val ak = new Sk_Ak()
        var lat_log = lat + "," + log
        //通过经纬度向百度云发送请求后获得商圈信息
        val addr: String = ak.getUserAddr(lat_log)
        if (addr.size != 0) {
          val base: String = GeoHash.withCharacterPrecision(lat.toDouble, log.toDouble, 8).toBase32
          redis.hset("SavaTag_user", base, addr)
        }
      }
    })
    redis.close()
  })
  sc.stop()
}
